package org.example


import org.apache.spark.sql.SparkSession

object zuoye {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()

    val sc = spark.sparkContext
    val filePath = "src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")

    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {
        (x(0), x(1), x(2))  }
         }.cache()
        println("平均得分最高的前 10 名的电影名称简单版") //  ( MovieId,( Rating, 1) )  (1200,(4.0,1))
      rating.map(x => (x._2, (x._3.toDouble, 1))) //  ( MovieId,( 总分, 总次数) )
      .reduceByKey((x, y) => {   (x._1 + y._1, x._2 + y._2)  }) // (  平均分, MovieId )
      .map(x => (x._2._1 / x._2._2, x._1))
        .sortByKey(false)
        .take(5)
        .foreach(println)
        println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")

    val moviesInfo = moviesRDD.map(x => x.split("::"))
    .map(x => {
      (x(0), (x(1), x(2)))

    })

    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1))) //  ( MovieId,( Rating, 1) ) (1252,(4.0,1))
     .reduceByKey((x, y) => {  (x._1 + y._1, x._2 + y._2)  }) //  ( MovieId,( 总分, 总次数) )
    .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))//   ( MovieId, (平均分, 总分,总次数) )
    moviesInfo.join(ratingsInfo)
    .map(info => {
      (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
       // (平均分,(movieId,Title,Genres,总分,总次数))
      }).sortByKey(false)   .take(5)   .foreach(println)
      println("观影人数最多的前 10 部电影")

    val watchViewsInfo = rating.map(x => {
      (x._2, 1)

    }).reduceByKey((x, y) => x + y) //  ( MovieId,总次数 )
      .map(x => (x._2, x._1))
      .sortByKey(false)  .take(5) // 5 名
    watchViewsInfo.foreach(println(_))
    println("===================>")
    rating.map(x => (x._2, 1)) //  ( MovieId, 1)
     .reduceByKey((x, y) => {   (x + y)
     }) //  ( MovieId,总次数 )
      .map(x => (x._2, x._1)) //  (  总次数, MovieId )
     .sortByKey(false)
      .take(10)
      .foreach(println) //  286-> 999
     println("详情的输出(  观影人数，电影编号)")
    moviesInfo.join(ratingsInfo).map(x => {   (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))   })
      .sortByKey(false)   .take(10)   .foreach(println)

  }
}
